package chen.bupt.mapreduce.board.semantic;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.util.HDFSFileUtils;

public class SynonymsFeatureJob {
	private static String inputBase = "/user/chenlingpeng/smth/board_feature/";
	private static String outputBase = "/user/chenlingpeng/smth/board_feature_synonyms/";
	private static final String synonymsPath = "/user/chenlingpeng/smth/synonyms.txt";

	private static class WordCount {
		public String word;
		public int count;

		public WordCount(String s, int c) {
			word = s;
			count = c;
		}
		
		public String toString(){
			return word+":"+count;
		}
	}

	private static class MyCompatator implements Comparator<WordCount> {

		@Override
		public int compare(WordCount w1, WordCount w2) {
			// TODO Auto-generated method stub
			if (w1.count == w2.count)
				return w1.word.compareTo(w2.word);
			return w1.count - w2.count;
		}

	}

	public static class SynonymsTFIDFMap extends
			Mapper<LongWritable, Text, Text, NullWritable> {
		private Map<String, String> synMap = new HashMap<String, String>();

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			List<String> synonyms = HDFSFileUtils.readLines(synonymsPath);
			for (String s : synonyms) {
				synMap.put(s.split(",")[1], s.split(",")[0]);
			}
		}

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			if (value.toString().startsWith("total"))
				return;
			String[] val = value.toString().split(";");
			Map<String, Integer> tmpMap = new HashMap<String, Integer>();
			for (int i = 1; i < val.length; i++) {
				String[] tmp = val[i].split(":");
				if (synMap.containsKey(tmp[0])) {
					tmp[0] = synMap.get(tmp[0]);
				}
				Integer d = tmpMap.get(tmp[0]);
				if (d == null) {
					tmpMap.put(tmp[0], Integer.parseInt(tmp[1]));
				} else {
					tmpMap.put(tmp[0], d + Integer.parseInt(tmp[1]));
				}
			}
			StringBuffer sb = new StringBuffer();
			Set<String> set = tmpMap.keySet();
			List<WordCount> list = new ArrayList<WordCount>();
			
			for (String k : set) {
				list.add(new WordCount(k,tmpMap.get(k)));
			}
			Collections.sort(list, new MyCompatator());
			for(WordCount wc:list){
				sb.append(wc.toString()+";");
			}
			
			context.write(new Text(sb.toString()), NullWritable.get());
			context.write(new Text("total size: " + set.size()), NullWritable
					.get());
			
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		Configuration confg = new Configuration();
		confg.set("fs.default.name", "hdfs://10.1.1.26:31054");
		FileSystem fs = FileSystem.get(confg);
		FileStatus[] fsDirs = fs.listStatus(new Path(inputBase));
		for (FileStatus fss : fsDirs) {
			if (!fss.isDir())
				continue;
			Path dirName = fss.getPath();
			String boardId = dirName.getName();
			if (boardId.startsWith("_"))
				continue;
			if (!boardId.startsWith(args[0]))
				continue;
			Configuration conf = new Configuration();
			Job job = new Job(conf, "SynonymsFeatureJob_"+boardId);
			job.setJarByClass(SynonymsTFIDFJob.class);
			job.setMapperClass(SynonymsTFIDFMap.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(NullWritable.class);
			Path input = new Path(inputBase + boardId);
			Path output = new Path(outputBase + boardId);
			HDFSFileUtils.deleteFile(output, conf);
			FileInputFormat.addInputPath(job, input);
			FileOutputFormat.setOutputPath(job, output);
			job.waitForCompletion(true);
		}
	}

}
